package com.atguigu.flink.sql.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/11
 *  sql中不支持计数窗口。
 *
 *      Group Window Functions: 常规的 window聚合
 *                                  提供了 滚动，会话，滑动。
 *
 *                   TVF window:  比Group Window更有优势
 *                          1.可以进行性能优化
 *                          2.支持标准的 GROUPING SETS
 *                          3.支持topN
 *
 *
 *
 *
 */
public class Demo5_TVFWindowGroupingSets
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        //自带水印，自带eventtime
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                             .column("id", "STRING")
                             .column("ts", "BIGINT")
                             .column("vc", "INT")
                             .columnByExpression("pt", "proctime()")
                             .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .watermark("et","et - INTERVAL '0.001' SECOND")
                             .build();
        //从流中获取时间属性
        Table table = tableEnvironment.fromDataStream(ds,schema);

        //为表起名字
        tableEnvironment.createTemporaryView("t1",table);



        String tumbleWindowSql = "SELECT id, window_start, window_end, SUM(vc) sumVc" +
            "  FROM TABLE(" +
            "    TUMBLE(TABLE t1, DESCRIPTOR(et), INTERVAL '5' SECOND ))" +
            "  GROUP BY window_start, window_end , " +
            //"  grouping sets ( (id), () ) "
            //"  cube( id ) "
            " rollup( id ) "
            ;


        //使用sql操作
        tableEnvironment.sqlQuery(tumbleWindowSql)
                        .execute().print();

        env.execute();


    }
}
